package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.function.*;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.sink.SinkHbase;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 指数秒级行情
 * 每5s一条最新时间的数据
 */
public class IndexSecTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {
        /**
         * 开发步骤：
         * 1.数据分组
         * 2.划分时间窗口
         * 3.秒级数据处理（新建数据写入Bean和秒级窗口函数）
         * 4.数据写入操作
         *   * 封装ListPuts
         *   * 数据写入
         */
        //1.数据分组
        waterData
                .keyBy(new KeyFunction())
                //2.划分时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //4.秒级窗口函数业务处理
                .apply(new IndexSecWindowFunction())
                //5.数据写入操作
                //封装ListPuts
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5))) //不分组,获取所有的最新个股数据
                .apply(new IndexSecListPutWindowFunction())
                //数据写入
                .addSink(new SinkHbase(QuotConfig.config.getProperty("index.hbase.table.name")));

    }
}
